use crate::imp::ws::websocket_imp::MessageData;
use crate::imp::ws::websocket_imp::SendData;
use crate::imp::ws::websocket_imp::WsReceivedMessage;
use crate::imp::ws::websocket_imp::WsSendMessage;
use crate::utils::ids;
use crate::utils::mpsc_utils::get_sender_receive;
use crate::utils::mpsc_utils::send_with_cache;
use async_trait::async_trait;
use futures_channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
use futures_util::{AsyncWriteExt, SinkExt};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;
use std::time::Duration;

#[async_trait]
pub trait TunnelSignalTrait: Sync + Send {
    async fn send(
        &mut self,
        target_id: String,
        event: String,
        data: String,
    ) -> Result<(), anyhow::Error>;
}

#[derive(Clone)]
pub struct SignalAdapter {
    receiver_tx: UnboundedSender<WsReceivedMessage>,
    signal_send_map: Arc<RwLock<HashMap<String, (String, String)>>>,
}

impl SignalAdapter {
    pub fn insert_send_msg(&self, event: String, data: String) -> String {
        let event_id = ids::id_generate_xbit(6);
        {
            let mut rw_lock_map = (*self.signal_send_map).write().unwrap();
            rw_lock_map.insert(event_id.to_owned(), (event, data));
            log::info!(
                "signal adapt insert rw_lock_map size = {}",
                rw_lock_map.len()
            );
        }
        event_id
    }

    pub fn remove_send_msg(&self, send_id: String) -> Option<(String, String)> {
        let mut remove_msg = None;
        {
            let mut rw_lock_map = (*self.signal_send_map).write().unwrap();
            let item = rw_lock_map.remove(&send_id);
            log::info!(
                "signal adapt remove rw_lock_map size = {}",
                rw_lock_map.len()
            );
            remove_msg = item;
        }
        remove_msg
    }

    pub async fn create(
        events: Vec<String>,
    ) -> Result<(Self, UnboundedReceiver<WsReceivedMessage>), anyhow::Error> {
        // unbounded receiver_tx and receive_rx.
        let (receiver_tx, receive_rx) = mpsc::unbounded();
        // loop and add recieve signal message sender to websocket.
        let adapte_rec_tx = receiver_tx.clone();
        tokio::spawn(async move {
            let mut receive = get_sender_receive();
            loop {
                let sender = receive.borrow_and_update().clone();
                match sender {
                    Some(mut sender) => {
                        // add UnboundedReceiver<WsReceivedMessage> to websocket
                        if let Err(e) = sender
                            .send(WsSendMessage::AddReceiver(
                                events.to_owned(),
                                receiver_tx.clone(),
                            ))
                            .await
                        {
                            log::warn!("WsSendMessage::AddReceiver err: {:?}", e);
                        }
                    }
                    None => {}
                }
                if let Err(e) = receive.changed().await {
                    log::warn!("tunnel SignalAdapter receive.changed() err: {:?}", e);
                }
            }
        });

        Ok((
            Self {
                receiver_tx: adapte_rec_tx,
                signal_send_map: Arc::new(RwLock::new(HashMap::new())),
            },
            receive_rx,
        ))
    }
}

#[async_trait]
impl TunnelSignalTrait for SignalAdapter {
    async fn send(
        &mut self,
        target_id: String,
        event: String,
        data: String,
    ) -> Result<(), anyhow::Error> {
        let event_id = self.insert_send_msg(event.to_owned(), data.to_owned());
        send_with_cache(
            WsSendMessage::MessageToSend(SendData {
                target_id,
                msg: MessageData {
                    event,
                    data,
                    id: event_id,
                },
            }),
            Duration::from_secs(8),
        )?;
        Ok(())
    }
}
